package com.yupi.springbootinit.kafka.consumer;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.yupi.springbootinit.exception.BusinessException;
import com.yupi.springbootinit.kafka.contast.KafkaTopicConstant;
import com.yupi.springbootinit.mapper.ChatSessionsMapper;
import com.yupi.springbootinit.model.entity.ChatSessions;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.util.Optional;

@Component
@Slf4j
public class MessageConsumer {

    @Resource
    private ChatSessionsMapper chatSessionsMapper;

    @KafkaListener(topics = KafkaTopicConstant.TOPIC_ADD, groupId = "ADD_GROUP", concurrency = "1", containerFactory = "kafkaManualAckListenerContainerFactory")
    public void MysqlAdd(ConsumerRecord<?, ?> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
        Optional.ofNullable(record.value()).ifPresent(msg -> {
            if (msg instanceof ChatSessions) {
                ChatSessions chatSession = (ChatSessions) msg;
                try {
                    // 通过 sessionId 查询是否已经存在
                    ChatSessions existingSession = chatSessionsMapper.selectBySessionId(chatSession.getSessionId());
                    System.out.println("existingSession: " + existingSession);
                    if (existingSession != null) {
                        // 更新已有的 session
                        existingSession.setMessageText(chatSession.getMessageText());
                        LambdaQueryWrapper<ChatSessions> wrapper = new LambdaQueryWrapper<>();
                        System.out.println("session_id: " + existingSession.getSessionId());
                        wrapper.eq(ChatSessions::getSessionId, existingSession.getSessionId());
                        chatSessionsMapper.update(existingSession, wrapper);
                    } else {
                        // 插入新的 session
                        chatSessionsMapper.insert(chatSession);
                    }
                    // 手动ACK
                    ack.acknowledge();
                    log.info("数据同步添加成功! Topic: {}, Message: {}", topic, chatSession);
                } catch (Exception e) {
                    log.error("数据添加失败！Topic: {}, Message: {}", topic, chatSession, e);
                    throw new BusinessException(500, "数据添加失败");
                }
            } else {
                log.error("接收到的消息不是 ChatSessions 类型: {}", msg);
            }
        });
    }
}
